/**                                                                                                                                                                                
 * Copyright (c) 2010 Yahoo! Inc. All rights reserved.                                                                                                                             
 *                                                                                                                                                                                 
 * Licensed under the Apache License, Version 2.0 (the "License"); you                                                                                                             
 * may not use this file except in compliance with the License. You                                                                                                                
 * may obtain a copy of the License at                                                                                                                                             
 *                                                                                                                                                                                 
 * http://www.apache.org/licenses/LICENSE-2.0                                                                                                                                      
 *                                                                                                                                                                                 
 * Unless required by applicable law or agreed to in writing, software                                                                                                             
 * distributed under the License is distributed on an "AS IS" BASIS,                                                                                                               
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or                                                                                                                 
 * implied. See the License for the specific language governing                                                                                                                    
 * permissions and limitations under the License. See accompanying                                                                                                                 
 * LICENSE file.                                                                                                                                                                   
 */

package com.yahoo.ycsb;

import java.util.HashMap;
import java.util.Properties;
import java.util.Set;
import java.util.Enumeration;
import java.util.Vector;

import com.yahoo.ycsb.DBException;
import com.yahoo.ycsb.ByteIterator;
import com.yahoo.ycsb.ByteArrayByteIterator;
import com.yahoo.ycsb.StringByteIterator;

/**
 * Basic DB that just prints out the requested operations, instead of doing them against a database.
 */
public class TeraDB extends DB
{
        public static final String VERBOSE="basicdb.verbose";
        public static final String VERBOSE_DEFAULT="false";
        
        public static final String SIMULATE_DELAY="basicdb.simulatedelay";
        public static final String SIMULATE_DELAY_DEFAULT="0";
        
        public static final String COLUMN_FAMILY="basicdb.columnfamily";
        public static final String COLUMN_FAMILY_DEFAULT="cf0";
        
        boolean verbose;
        int todelay;
        String column_family;

        public TeraDB()
        {
                todelay=0;
        }

        
        void delay()
        {
                if (todelay>0)
                {
                        try
                        {
                                Thread.sleep((long)Utils.random().nextInt(todelay));
                        }
                        catch (InterruptedException e)
                        {
                                //do nothing
                        }
                }
        }

        /**
         * Initialize any state for this DB.
         * Called once per DB instance; there is one DB instance per client thread.
         */
        @SuppressWarnings("unchecked")
        public void init()
        {
                verbose=Boolean.parseBoolean(getProperties().getProperty(VERBOSE, VERBOSE_DEFAULT));
                todelay=Integer.parseInt(getProperties().getProperty(SIMULATE_DELAY, SIMULATE_DELAY_DEFAULT));
                column_family=getProperties().getProperty(COLUMN_FAMILY, COLUMN_FAMILY_DEFAULT);
                
                if (verbose)
                {
                        System.out.println("***************** properties *****************");
                        Properties p=getProperties();
                        if (p!=null)
                        {
                                for (Enumeration e=p.propertyNames(); e.hasMoreElements(); )
                                {
                                        String k=(String)e.nextElement();
                                        System.out.println("\""+k+"\"=\""+p.getProperty(k)+"\"");
                                }
                        }
                        System.out.println("**********************************************");
                }
        }

        /**
         * Read a record from the database. Each field/value pair from the result will be stored in a HashMap.
         *
         * @param table The name of the table
         * @param key The record key of the record to read.
         * @param fields The list of fields to read, or null for all of them
         * @param result A HashMap of field/value pairs for the result
         * @return The result of the operation.
         */
        public Status read(String table, String key, Set<String> fields, HashMap<String,ByteIterator> result)
        {
                delay();

                if (fields!=null)
                {
                        String column = column_family + ":";
                        boolean first_field = true;
                        for (String f : fields)
                        {
                                if (!first_field)
                                {
                                        column += ",";
                                } else {
                                        column += f;
                                        first_field = false;
                                }
                        }
                        System.out.println("GET\t"+key+"\t"+column+"\t0");
                }
                else
                {
                        System.out.println("GET\t"+key+"\t"+column_family+"\t0");
                }

                return Status.OK;
        }

        /**
         * Perform a range scan for a set of records in the database. Each field/value pair from the result will be stored in a HashMap.
         *
         * @param table The name of the table
         * @param startkey The record key of the first record to read.
         * @param recordcount The number of records to read
         * @param fields The list of fields to read, or null for all of them
         * @param result A Vector of HashMaps, where each HashMap is a set field/value pairs for one record
         * @return The result of the operation.
         */
        public Status scan(String table, String startkey, int recordcount, Set<String> fields, Vector<HashMap<String,ByteIterator>> result)
        {
                delay();

                System.out.print("SCAN "+table+" "+startkey+" "+recordcount+" [ ");
                if (fields!=null)
                {
                        for (String f : fields)
                        {
                                System.out.print(f+" ");
                        }
                }
                else
                {
                        System.out.print("<all fields>");
                }

                System.out.println("]");

                return Status.OK;
        }

        /**
         * Update a record in the database. Any field/value pairs in the specified values HashMap will be written into the record with the specified
         * record key, overwriting any existing values with the same field name.
         *
         * @param table The name of the table
         * @param key The record key of the record to write.
         * @param values A HashMap of field/value pairs to update in the record
         * @return The result of the operation.
         */
        public Status update(String table, String key, HashMap<String,ByteIterator> values)
        {
                delay();

                if (values!=null)
                {
                        for (String k : values.keySet())
                        {
                                System.out.println("PUT\t"+key+"\t"+values.get(k)+"\t"+column_family+":"+k+"\t0");
                        }
                }

                return Status.OK;
        }

        /**
         * Insert a record in the database. Any field/value pairs in the specified values HashMap will be written into the record with the specified
         * record key.
         *
         * @param table The name of the table
         * @param key The record key of the record to insert.
         * @param values A HashMap of field/value pairs to insert in the record
         * @return The result of the operation.
         */
        public Status insert(String table, String key, HashMap<String,ByteIterator> values)
        {
                return update(table, key, values);
        }

        /**
         * Delete a record from the database. 
         *
         * @param table The name of the table
         * @param key The record key of the record to delete.
         * @return The result of the operation.
         */
        public Status delete(String table, String key)
        {
                delay();

                System.out.println("DELETE\t"+key);

                return Status.OK;
        }

}
